--user_session_stat => user_session_stat_rich
--2012-12-03
--jiankuiwang@meilishuo.com
SET job.priority VERY_HIGH;

register /hadoop/pig/lib/piggybank.jar 
register /hadoop/pig/lib/hive-exec-0.11.0.jar 
register /hadoop/pig/lib/hive-common-0.11.0.jar

DEFINE user_session_reborn_rich `user_session_reborn_rich.py` SHIP('user_session_reborn_rich.py','rule.conf');
DEFINE add_field `add_field.py` SHIP('add_field.py', 'IP.csv');
DEFINE UDFUnfoldURL `UDFUnfoldURL.py` SHIP('UDFUnfoldURL.py');

--载入原始数据 
LOG = LOAD '$HIVE_WAREHOUSE/visitlogs/dt=$date/*' USING org.apache.pig.piggybank.storage.HiveColumnarLoader('hour int,minute int,second int,class_name string,method_name string,mem_use int,run_time float,user_id int,agent string,refer string,type int,is_search_engine int,uri string,visitip string,poststr string,sessid string,channel_from int,tag string,http_code string,refer_class_name string,refer_method_name string,dt string,vhour string');

USS_YES = LOAD '$HIVE_WAREHOUSE/user_session_stat/dt=$yesterday/*' USING PigStorage('\t') AS (sessid:chararray,session_day:chararray,reg_day:chararray,userids:chararray,tags:chararray,visitips:chararray,land:chararray,refer:chararray,agent:chararray,session_pv:int,channel:chararray,sub_channel:chararray,is_spam:int,region:chararray,city:chararray,online_time:int,market_from:chararray,addTwitter:int,addGoods:int,delTwitter:int,voteTwitter:int,noteTwitter:int,addFollower:int,delFollower:int,sendMsg:int,addMagazine:int,twitterMagazine:int,folMagazine:int,quitMagazine:int,viewMsg:int,viewAct:int,viewWelfare:int,applyWelfare:int,clickTwitter:int,clickTaobao:int,clickCPC:int,clickCPS:int,clickHOT:int,clickBanner:int,clickHuodong:int,idc:chararray,viewList:int,clickP4P:int,dt:chararray);

NO200_LOG = LOAD '$HIVE_WAREHOUSE/visitlogs_no200/dt=$date/*' USING org.apache.pig.piggybank.storage.HiveColumnarLoader('hour int,minute int,second int,class_name string,method_name string,mem_use int,run_time float,user_id int,agent string,refer string,type int,is_search_engine int,uri string,visitip string,poststr string,sessid string,channel_from int,tag string,http_code string,refer_class_name string,refer_method_name string,dt string,vhour string');

--session_pv计算有效UV,join cmm是为了去除不必要的依赖
CMM = LOAD '$HIVE_WAREHOUSE/class_method_map/*' USING PigStorage('\u0001') AS (class:chararray, method:chararray, category:chararray);
SPAM =  LOAD '$HIVE_WAREHOUSE/spam_daily/dt=$date' USING PigStorage('\u0001') AS (type:int, op:int, value:chararray, algo:chararray);
FUT_GOODS_INFO = LOAD '$HIVE_WAREHOUSE/ods_dolphin_fut_goods_info/*' USING PigStorage('\u0001') AS (goods_id:int, goods_author_uid:int, goods_title:chararray, goods_url:long,goods_pic_url:chararray,goods_pic_id:int,goods_author_ctime:int,goods_source_type:chararray,goods_picture_id:int,goods_price:double,good_weight:int,cataid:int);

TAOBAO_COMMISSION = LOAD '$HIVE_WAREHOUSE/ods_dolphin_taobao_commission/*' USING PigStorage('\u0001') AS (t_num_iid:int,goods_id:int,t_commission:double,t_commission_rate:double,m_commission_rate:double,price:double,nick:chararray,ctime:chararray);
---------------------------------------载入原始日志结束
CPS_GOODS = FOREACH ( JOIN  FUT_GOODS_INFO BY goods_id , TAOBAO_COMMISSION BY goods_id ) GENERATE  FUT_GOODS_INFO::goods_id, goods_url;

CPS_VISITLOGS =  FOREACH (FILTER LOG BY SIZE(sessid)==32 and is_search_engine ==0 AND sessid != '' AND class_name != 'log' and (class_name == 'url' or class_name == 's' ) and (refer MATCHES '.*/share/.*' or refer MATCHES '.*/item/.*' )) GENERATE sessid, uri;
CPS_GOODS = DISTINCT CPS_GOODS;

CPS_GOODS_UNFOLDURL = STREAM CPS_VISITLOGS THROUGH UDFUnfoldURL AS (sessid:chararray, goods_url:long);
CPS_JOIN = FOREACH ( JOIN  CPS_GOODS_UNFOLDURL BY goods_url , CPS_GOODS BY goods_url ) GENERATE CPS_GOODS_UNFOLDURL::sessid, CPS_GOODS_UNFOLDURL::goods_url;

CPS_GROUP = GROUP CPS_JOIN BY sessid; 
FLAT_LOG_CPS = FOREACH CPS_GROUP GENERATE FLATTEN(group) as sessid,COUNT($1) as cps_count;
--------------------------------------CPS基础数据处理结束
VALID_SID = DISTINCT (FOREACH (FILTER SPAM BY type==0 and op==1) GENERATE value AS sessid); --sessid
NO200_LOG_VALID = FOREACH ( JOIN NO200_LOG BY sessid, VALID_SID BY sessid ) GENERATE NO200_LOG::sessid,is_search_engine,class_name,uri;

NO200_LOG_HOT = FOREACH (FILTER NO200_LOG_VALID BY SIZE(sessid)==32 and is_search_engine ==0 AND sessid != '' AND class_name != 'log' and uri MATCHES '/api/redirect/.*') GENERATE sessid, uri;

NO200_LOG_BANNER = FOREACH (FILTER NO200_LOG_VALID BY SIZE(sessid)==32 and is_search_engine ==0 AND sessid != '' AND class_name != 'log' and uri MATCHES '.*sid=.*mid=.*') GENERATE sessid, uri;

NO200_LOG_HOT_GROUP = GROUP NO200_LOG_HOT BY sessid;
FLAT_LOG_HOT = FOREACH NO200_LOG_HOT_GROUP GENERATE FLATTEN(group) as sessid,COUNT($1) as hot_count;

NO200_LOG_BANNER_GROUP = GROUP NO200_LOG_BANNER BY sessid;
FLAT_LOG_BANNER = FOREACH NO200_LOG_BANNER_GROUP GENERATE FLATTEN(group) as sessid,COUNT($1) as banner_count;
--------------------------------------NO200日志处理结束
-- 过滤，拼装成基础日志
LOG_SMALL = FOREACH (FILTER LOG BY SIZE(sessid)==32 and is_search_engine ==0 AND sessid != '' AND class_name != 'log') GENERATE sessid, user_id, visitip, uri, refer, agent, tag, hour, minute, second, CONCAT(class_name,method_name) as JOIN_KEY, class_name, method_name; --13 在前面过滤掉

-- 增加is_valid字段标识是否当行是一个有效PV
LOG_VALID_CMM = FOREACH ( JOIN  LOG_SMALL BY JOIN_KEY LEFT OUTER, CMM BY CONCAT(class, method) ) GENERATE sessid, user_id, visitip, uri, refer, agent, tag, hour, minute, second ,((CMM::class is null) ? 0 : 1) AS is_valid:int, LOG_SMALL::class_name, LOG_SMALL::method_name ; --13  join合并class_name,method_name成一个key

--增加is_spam字段,增加is_meilishuo标记meilishuo和非mls的refer,处理并发请求
VALID_SID = DISTINCT (FOREACH (FILTER SPAM BY type==0 and op==1) GENERATE value AS sessid); --sessid
LOG_VALID = FOREACH ( JOIN LOG_VALID_CMM BY sessid LEFT OUTER, VALID_SID BY sessid ) GENERATE LOG_VALID_CMM::LOG_SMALL::sessid, user_id, visitip, uri, refer, agent, tag, hour, minute, second ,is_valid, ((refer matches '^http://www.meilishuo.com.*') ? 1 : 0) as is_meilishuo, ((VALID_SID::sessid is null) ? 1 : 0) AS is_spam, class_name, method_name;--15  1代表是spam,0代表非spam

--增加 region, city字段
LOG_VALID_REGION = STREAM LOG_VALID THROUGH add_field AS (sessid:chararray, user_id:int, visitip:chararray, uri:chararray, refer:chararray, agent:chararray, tag:chararray, hour:int, minute:int, second:int, is_valid:int, is_meilishuo:int, is_spam:int, class_name:chararray, method_name:chararray, region:chararray, city:chararray,idc:chararray); --18

-----------------------------------VISITLOG拼接结束

LOG_VALID_HOT = FOREACH ( JOIN  LOG_VALID_REGION BY sessid LEFT OUTER, FLAT_LOG_HOT BY sessid ) GENERATE LOG_VALID_REGION::sessid, user_id, visitip, uri, refer, agent, tag, hour, minute, second, is_valid, is_meilishuo, is_spam, class_name, method_name, region, city, idc, (( FLAT_LOG_HOT::sessid is null) ? 0 : FLAT_LOG_HOT::hot_count) AS hot_count;

LOG_VALID_BANNER = FOREACH ( JOIN  LOG_VALID_HOT BY sessid LEFT OUTER, FLAT_LOG_BANNER BY sessid ) GENERATE LOG_VALID_HOT::LOG_VALID_REGION::sessid, user_id, visitip, uri, refer, agent, tag, hour, minute, second, is_valid, is_meilishuo, is_spam, class_name, method_name, region, city, idc, hot_count, (( FLAT_LOG_BANNER::sessid is null) ? 0 : FLAT_LOG_BANNER::banner_count) AS banner_count;--20

-----------------------------------NO200 和 VISITLOGS 合并
GROUP_LOG = GROUP LOG_VALID_BANNER by sessid;

FLAT_LOG = FOREACH GROUP_LOG {ORDER_JOIN_LOG = ORDER LOG_VALID_BANNER BY hour, minute, second, is_meilishuo;GENERATE FLATTEN(ORDER_JOIN_LOG);};

--LIMIT_LOG = LIMIT LOG_VALID 100;
--DUMP LIMIT_LOG;

--开始数据数据 调用user_session_reborn.py 格式化数据
USER_SESSION_ROWS = STREAM FLAT_LOG THROUGH user_session_reborn_rich AS (sessid:chararray,session_day:chararray,reg_day:chararray,userids:chararray,tags:chararray,visitips:chararray,land:chararray,refer:chararray,agent:chararray,session_pv:int,channel:chararray,sub_channel:chararray,is_spam:int,region:chararray,city:chararray,idc:chararray,online_time:int,market_from:chararray,addTwitter:int,addGoods:int,delTwitter:int,voteTwitter:int,noteTwitter:int,addFollower:int,delFollower:int,sendMeg:int,addMagazine:int,twitterMagazine:int,folMagazine:int,quitMagazine:int,viewMsg:int,viewAct:int,viewWelfare:int,applyWelfare:int,clickTwitter:int,clickTaobao:int,clickCPC:int,clickCPS:int,clickHOT:int,clickBanner:int,clickHuodong:int,viewList:int,clickP4P:int); --42

USER_SESSION_ROWS = FOREACH ( JOIN  USER_SESSION_ROWS BY sessid LEFT OUTER, FLAT_LOG_CPS BY sessid ) GENERATE USER_SESSION_ROWS::sessid, session_day, reg_day, userids, tags, visitips, land, refer, agent, session_pv, channel, sub_channel,is_spam,region,city,online_time,market_from,addTwitter,addGoods,delTwitter,voteTwitter,noteTwitter,addFollower,delFollower,sendMeg,addMagazine,twitterMagazine,folMagazine,quitMagazine,viewMsg,viewAct,viewWelfare,applyWelfare,clickTwitter,clickTaobao,clickCPC,((FLAT_LOG_CPS::cps_count is null) ? 0 : FLAT_LOG_CPS::cps_count) as clickCPS, clickHOT,clickBanner,clickHuodong,idc,viewList,clickP4P; 

USS_YES = FOREACH ( FILTER USS_YES BY is_spam == 0 and session_pv > 0 ) GENERATE sessid, channel, sub_channel;
USER_SESSION_ROWS_RICH = FOREACH ( JOIN USER_SESSION_ROWS by sessid LEFT OUTER , USS_YES by sessid ) GENERATE USER_SESSION_ROWS::sessid, session_day, reg_day, userids, tags, visitips, land, refer, agent, session_pv, ( (USER_SESSION_ROWS::sub_channel=='meilishuo' and USS_YES::sessid is not null) ? USS_YES::channel : USER_SESSION_ROWS::channel ), ( (USER_SESSION_ROWS::sub_channel == 'meilishuo' and USS_YES::sessid is not null) ? USS_YES::sub_channel : USER_SESSION_ROWS::sub_channel ), is_spam, region, city, online_time, market_from, addTwitter,addGoods,delTwitter,voteTwitter,noteTwitter,addFollower,delFollower,sendMeg, addMagazine,twitterMagazine,folMagazine,quitMagazine,viewMsg,viewAct,viewWelfare,applyWelfare,clickTwitter,clickTaobao,clickCPC, clickCPS, clickHOT,clickBanner,clickHuodong,idc,viewList,clickP4P ;   

--DUMP USER_SESSION_ROWS;

STORE USER_SESSION_ROWS_RICH INTO '$HDFS_TMP_HOME/$store_name';
